#include <iostream>
#include <fstream>
#include <memory>
#include <thread>
#include <chrono>
#include <atomic>
using namespace std;

#include "qmdb.monitor.pb.h"
#include <google/protobuf/io/zero_copy_stream.h>
#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
#include <google/protobuf/io/gzip_stream.h>
#include <WinSock2.h>

#ifdef _DEBUG_
#pragma comment(lib, "libprotobufd.lib")
#pragma comment(lib, "libprotocd.lib")
#pragma comment(lib, "zlibd.lib")
#else
#pragma comment(lib, "libprotobuf.lib")
#pragma comment(lib, "libprotoc.lib")
#pragma comment(lib, "zlib.lib")
#endif //_DEBUG_

#pragma comment(lib, "Ws2_32.lib")

void printmsg(const qmdb::monitor& msg)
{
    cout << "id:" << msg.id() << endl;
    cout << "cmdcode:" << msg.cmdcode() << endl;
    cout << "msg:" << msg.msg() << endl;

    cout << "====repeat information start====" << endl;
    int repeatsize = msg.testrepeat_size();
    cout << "testrepeat_size:" << repeatsize << endl;
    for (int i = 0; i < repeatsize; i++)
    {
        cout << "testrepeat" << i << ":" << msg.testrepeat(i) << endl;
    }
    cout << "====repeat information end======" << endl;
}

atomic<bool> ready(false);
atomic<bool> tostop(false);

void service(int testTimes)
{
    sockaddr_in clientService;
    clientService.sin_family = AF_INET;
    clientService.sin_addr.s_addr = INADDR_ANY;
    clientService.sin_port = htons(9000);

    SOCKET ConnectSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (INVALID_SOCKET == ConnectSocket)
    {
        cerr << "invalid socket." << __LINE__ << endl;
        return;
    }

    if (::bind(ConnectSocket, (sockaddr*)&clientService, sizeof(clientService)) == -1)
    {
        cerr << "bind failed, errno:" << errno << endl;
        return;
    }

    if (::listen(ConnectSocket, 10) == -1)
    {
        cerr << "listen failed, errno:" << errno << endl;
        return;
    }

    ready.store(true);

    SOCKET clientSocket = INVALID_SOCKET;
    int ilen = sizeof(clientService);
    if ((clientSocket = ::accept(ConnectSocket, (sockaddr*)(&clientService), &ilen)) == INVALID_SOCKET)
    {
        cerr << "accept invalid socket, errno:" << errno << endl;
        return;
    }

    vector<char> recvBuff(1024, 0);
    unique_ptr<char[]> gzipbuff(new char[1024]);
    memset(gzipbuff.get(), 0, 1024);
    int recvMsg = 0;
    int currentPos = 0;
    while (!tostop.load())
    {
        int totalCount = 0;
        int readCount = 0;
        const int headLen = 4;
        google::protobuf::uint32 bodyLen = 0;
        while ((readCount = ::recv(clientSocket, (char*)(&*recvBuff.begin())+totalCount, recvBuff.size() - 1 - totalCount, 0)) > 0)
        {
            totalCount += readCount;
            if (totalCount < 4)
            {
                continue;
            }

            if (bodyLen == 0 && totalCount >= 4)
            {
                google::protobuf::io::ArrayInputStream arrayInputStream((char*)(&*recvBuff.begin()), headLen);
                google::protobuf::io::CodedInputStream codedInputStream(&arrayInputStream);
                codedInputStream.ReadVarint32(&bodyLen);
                currentPos = codedInputStream.CurrentPosition();
            }

            if (recvBuff.size() < (int)bodyLen + currentPos)
            {
                recvBuff.resize((int)bodyLen + currentPos);
                cout << "resize" << endl;
            }

            while (totalCount >= (int)bodyLen + currentPos)
            {
                google::protobuf::io::ArrayInputStream arrayInputStream(((char*)(&*recvBuff.begin()))+ currentPos, bodyLen);
                google::protobuf::io::CodedInputStream codedInputStream(&arrayInputStream);
                codedInputStream.ReadRaw(gzipbuff.get(), bodyLen);
                google::protobuf::io::ArrayInputStream array_gzip_in(gzipbuff.get(), bodyLen);
                google::protobuf::io::GzipInputStream gzip_in(&array_gzip_in);
                qmdb::monitor msg1;
                if (!msg1.ParseFromZeroCopyStream(&gzip_in))
                {
                    cerr << "parse failed." << recvMsg << endl;
                    cerr << "totalCount:" << totalCount << " bodyLen:" << bodyLen << endl;
                    cerr << "body:" << endl;
                    for (int i = currentPos; i < currentPos + bodyLen; i++)
                    {
                        cout << std::hex << (unsigned int)(recvBuff[i]);
                    }
                    cout << std::dec <<endl;
                    exit( - 1);
                }

                if (totalCount > (int)bodyLen + currentPos)
                {
                    std::copy(recvBuff.begin() + (int)bodyLen + currentPos, recvBuff.begin() + totalCount, recvBuff.begin());
                }

                totalCount -= ((int)bodyLen + currentPos);
                bodyLen = 0;
                if (totalCount >= headLen)
                {
                    google::protobuf::io::ArrayInputStream arrayInputStream((char*)(&*recvBuff.begin()), headLen);
                    google::protobuf::io::CodedInputStream codedInputStream(&arrayInputStream);
                    codedInputStream.ReadVarint32(&bodyLen);
                    currentPos = codedInputStream.CurrentPosition();
                }
                recvMsg++;
            }

            if (recvMsg >= testTimes)
            {
                break;
            }
        }
    }

    cout << "thread exit ......" << endl;
    cout << "get message count:" << recvMsg << endl;
}

int main(int argc, char* argv[])
{
    WSADATA wsaData;
    int iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
    if (iResult != NO_ERROR) {
        cerr << "WSAStartup function failed with error: " << iResult << endl;
        return 1;
    }

    qmdb::msghead header;
    header.set_type(1);
    header.set_bodylen(2);
    cout << "header.ByteSize:" << header.ByteSize() << endl;

    char headbuff[8] = { 0 };
    google::protobuf::io::ArrayOutputStream array_head(headbuff, sizeof(headbuff));
    google::protobuf::io::CodedOutputStream coded_head(&array_head);
    header.SerializeToCodedStream(&coded_head);
    cout << "coded_head.ByteCount:" << coded_head.ByteCount() << endl;
    
    for (int i = 0; i < coded_head.ByteCount(); i++)
    {
        cout << std::hex << (unsigned int)headbuff[i];
    }
    cout << std::dec << endl;

    qmdb::monitor msg1;
    const int testTimes = 10000;
    msg1.set_id(1);
    msg1.set_cmdcode(200);
    msg1.set_msg("hello world!");
    for (int i = 0; i < 10; i++)
    {
        msg1.add_testrepeat("hello1");
        google::protobuf::MapPair<google::protobuf::int32, string> toInsert1(i, "abc");
        msg1.mutable_testmap()->insert(toInsert1);
    }
    
    unique_ptr<char[]> gzipouput(new char[1024]);
    unique_ptr<char[]> testbuff(new char[1024]);
    google::protobuf::io::ArrayOutputStream array_test(testbuff.get(), 1024);
    msg1.SerializeToZeroCopyStream(&array_test);
    cout << "msg1.ByteSize:" << msg1.ByteSize() << " array_test.ByteCount:" << array_test.ByteCount() << endl;
    google::protobuf::io::ArrayOutputStream array_test1(testbuff.get(), 1024);
    google::protobuf::io::CodedOutputStream coded_test1(&array_test1);
    msg1.SerializeToCodedStream(&coded_test1);
    cout << "msg1.ByteSize:" << msg1.ByteSize() << " coded_test1.ByteCount:" << coded_test1.ByteCount() << endl;
    
    google::protobuf::io::ArrayOutputStream array_output(gzipouput.get(), 1024);
    cout << "array_output.ByteCount():" << array_output.ByteCount() << endl;
    google::protobuf::io::GzipOutputStream gzip_output(&array_output);
    msg1.SerializeToZeroCopyStream(&gzip_output);
    cout << "array_output.ByteCount():" << array_output.ByteCount() << endl;
    gzip_output.Flush();
    cout << "array_output.ByteCount():" << array_output.ByteCount() << endl;
    gzip_output.Close();
    cout << "array_output.ByteCount():" << array_output.ByteCount() << endl;

    google::protobuf::io::ArrayInputStream array_input(gzipouput.get(), array_output.ByteCount());
    google::protobuf::io::GzipInputStream gzip_input(&array_input);
    qmdb::monitor msgxx;
    msgxx.ParseFromZeroCopyStream(&gzip_input);
    cout << "XXXXXXXXXXXXXXXXXXXXXX" << endl;
    cout << msgxx.ShortDebugString() << endl;
    return 0;
    printmsg(msgxx);
    for (int i = 0; i < array_output.ByteCount(); i++)
    {
        cout << std::hex <<(unsigned int)gzipouput[i];
    }
    cout << endl;

    google::protobuf::io::ArrayOutputStream array_(testbuff.get(), 1024);
    google::protobuf::io::CodedOutputStream coded_(&array_);
    coded_.WriteVarint32(array_output.ByteCount());
    cout << "coded_.ByteCount():" << coded_.ByteCount() << endl;
    coded_.WriteRaw(gzipouput.get(), array_output.ByteCount());
    cout << "coded_.ByteCount():" << coded_.ByteCount() << " array_output.ByteCount():" << array_output.ByteCount() << endl;
    return 0;
    for (int i = 0; i < coded_.ByteCount(); i++)
    {
        cout << std::hex << (unsigned int)testbuff[i];
    }
    cout << std::dec <<endl;

    google::protobuf::io::ArrayInputStream array_in(testbuff.get(), coded_.ByteCount());
    google::protobuf::io::CodedInputStream coded_in(&array_in);
    google::protobuf::uint32 dataLen = 0;
    coded_in.ReadVarint32(&dataLen);
    cout << "dataLen:" << dataLen << endl;
    memset(gzipouput.get(), 0, 1024);
    coded_in.ReadRaw(gzipouput.get(), dataLen);
    google::protobuf::io::ArrayInputStream array_gzip_in(gzipouput.get(), dataLen);
    google::protobuf::io::GzipInputStream gzip_in(&array_gzip_in);
    qmdb::monitor msgfromzip;
    msgfromzip.ParseFromZeroCopyStream(&gzip_in);
    cout << "msgfromzip:" << endl;
    printmsg(msgfromzip);

    thread servthread(service, testTimes);

    string serializedString = msg1.SerializeAsString();
    cout << "serialized string size=" << serializedString.size() << " content:\n" << serializedString << endl;
    cout << "bytesize:" << msg1.ByteSize() << endl;
    qmdb::monitor msg2;
    msg2.ParseFromString(serializedString);
    printmsg(msg2);

    SOCKET ConnectSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (INVALID_SOCKET == ConnectSocket)
    {
        cerr << "invalid socket." << __LINE__ << endl;
        return -1;
    }

    while (!ready.load())
    {
        this_thread::sleep_for(chrono::seconds(1));
        cout << "waiting for thread ready ..." << endl;
    }

    sockaddr_in clientService;
    clientService.sin_family = AF_INET;
    clientService.sin_addr.s_addr = inet_addr("127.0.0.1");
    clientService.sin_port = htons(9000);
    iResult = connect(ConnectSocket, (SOCKADDR *)& clientService, sizeof(clientService));
    if (iResult == SOCKET_ERROR) 
    {
        cerr << "connect function failed with error: " << errno << endl;
        iResult = closesocket(ConnectSocket);
        if (iResult == SOCKET_ERROR)
            cerr << "closesocket function failed with error: " << errno << endl;
        WSACleanup();
        return 1;
    }

    int sendMsg = 0;
    for (int i=0; i<testTimes; i++)
    {
        int sendbytes = send(ConnectSocket, testbuff.get(), coded_.ByteCount(), 0);
        if (-1 == sendbytes)
        {
            cerr << "send failed errno:" << errno << " errmsg:" << strerror(errno) << endl;
            return -1;
        }
        sendMsg++;
    }

    cout << "send message count:" << sendMsg << endl;
    //this_thread::sleep_for(chrono::seconds(1));
    tostop.store(true);
    //this_thread::sleep_for(chrono::seconds(5));
    //closesocket(ConnectSocket);
    //ConnectSocket = INVALID_SOCKET;
    servthread.join();
    cout << "exit ..." << endl;

    /*cout << "+++++++++++++++++++++++++++++++" << endl;
    fstream output("./log", ios::out | ios::trunc | ios::binary);
    if (!msg1.SerializeToOstream(&output))
    {
        cerr << "Failed to write msg." << endl;
        return -1;
    }
    output.close();

    qmdb::monitor msg3;
    fstream input("./log", ios::in | ios::binary);
    if (!msg3.ParseFromIstream(&input))
    {
        cerr << "Failed to parse msg." << endl;
        return -1;
    }
    input.close();
    printmsg(msg3);*/
    WSACleanup();

    return 0;
}